package com.example.demo.controller;

import com.example.demo.componet.MapUtil;
import com.example.demo.componet.ReadCSV;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import scala.Tuple2;

import javax.annotation.Resource;
import java.util.*;

@RestController
@RequestMapping("/analyze")
public class PublicProduct {

    @Resource
    private JavaSparkContext sc;

    //文件地址
    @Autowired
    private String getFilePath;

    private List<ArrayList> getDataList() {

        ReadCSV readCSV = new ReadCSV();
        //List<ArrayList> dataList =
        return  readCSV.readYearAnyColumn(getFilePath, 5);    //"C:\\Users\\fireapproval\\Desktop\\毕业设计\\数据集\\test.csv"

    }
    @GetMapping("/publicProduct")
    public ArrayList<ArrayList> method1(){

        //System.out.println(getDataList());

        //一月商品列表
        ArrayList<String> productList1 = new ArrayList<>();
        ArrayList<String> productList2 = new ArrayList<>();
        ArrayList<String> productList3 = new ArrayList<>();
        ArrayList<String> productList4 = new ArrayList<>();
        ArrayList<String> productList5 = new ArrayList<>();
        ArrayList<String> productList6 = new ArrayList<>();
        ArrayList<String> productList7 = new ArrayList<>();
        ArrayList<String> productList8 = new ArrayList<>();
        ArrayList<String> productList9 = new ArrayList<>();
        ArrayList<String> productList10 = new ArrayList<>();
        ArrayList<String> productList11 = new ArrayList<>();
        ArrayList<String> productList12 = new ArrayList<>();



        for (ArrayList list:getDataList()
             ) {
            //月份
            Integer i  = Integer.parseInt((String) list.get(0));
            //商品
            String s = (String) list.get(1);
            if (i.intValue() == 1){
                productList1.add(s);
            }
            if (i.intValue() == 2){
                productList2.add(s);
            }
            if (i.intValue() == 3){
                productList3.add(s);
            }
            if (i.intValue() == 4){
                productList4.add(s);
            }
            if (i.intValue() == 5){
                productList5.add(s);
            }
            if (i.intValue() == 6){
                productList6.add(s);
            }
            if (i.intValue() == 7){
                productList7.add(s);
            }
            if (i.intValue() == 8){
                productList8.add(s);
            }
            if (i.intValue() == 9){
                productList9.add(s);
            }
            if (i.intValue() == 10){
                productList10.add(s);
            }
            if (i.intValue() == 11){
                productList11.add(s);
            }
            if (i.intValue() == 12){
                productList12.add(s);
            }
        }


        JavaRDD<String> one1 = sc.parallelize(productList1);
        JavaRDD<String> one2 = sc.parallelize(productList2);
        JavaRDD<String> one3= sc.parallelize(productList3);
        JavaRDD<String> one4 = sc.parallelize(productList4);
        JavaRDD<String> one5 = sc.parallelize(productList5);
        JavaRDD<String> one6 = sc.parallelize(productList6);
        JavaRDD<String> one7 = sc.parallelize(productList7);
        JavaRDD<String> one8 = sc.parallelize(productList8);
        JavaRDD<String> one9 = sc.parallelize(productList9);
        JavaRDD<String> one10 = sc.parallelize(productList10);
        JavaRDD<String> one11 = sc.parallelize(productList11);
        JavaRDD<String> one12 = sc.parallelize(productList12);


        JavaPairRDD<String, Integer> wordMap1 = one1.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1));
        JavaPairRDD<String, Integer> wordMap2 = one2.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1));
        JavaPairRDD<String, Integer> wordMap3 = one3.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1));
        JavaPairRDD<String, Integer> wordMap4 = one4.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1));
        JavaPairRDD<String, Integer> wordMap5 = one5.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1));
        JavaPairRDD<String, Integer> wordMap6 = one6.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1));
        JavaPairRDD<String, Integer> wordMap7 = one7.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1));
        JavaPairRDD<String, Integer> wordMap8 = one8.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1));
        JavaPairRDD<String, Integer> wordMap9 = one9.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1));
        JavaPairRDD<String, Integer> wordMap10 = one10.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1));
        JavaPairRDD<String, Integer> wordMap11 = one11.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1));
        JavaPairRDD<String, Integer> wordMap12 = one12.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1));

        JavaPairRDD<String, Integer> reduceMap1 = wordMap1.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer+integer2);
        JavaPairRDD<String, Integer> reduceMap2 = wordMap2.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer+integer2);
        JavaPairRDD<String, Integer> reduceMap3 = wordMap3.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer+integer2);
        JavaPairRDD<String, Integer> reduceMap4 = wordMap4.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer+integer2);
        JavaPairRDD<String, Integer> reduceMap5 = wordMap5.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer+integer2);
        JavaPairRDD<String, Integer> reduceMap6 = wordMap6.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer+integer2);
        JavaPairRDD<String, Integer> reduceMap7 = wordMap7.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer+integer2);
        JavaPairRDD<String, Integer> reduceMap8 = wordMap8.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer+integer2);
        JavaPairRDD<String, Integer> reduceMap9 = wordMap9.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer+integer2);
        JavaPairRDD<String, Integer> reduceMap10 = wordMap10.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer+integer2);
        JavaPairRDD<String, Integer> reduceMap11 = wordMap11.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer+integer2);
        JavaPairRDD<String, Integer> reduceMap12 = wordMap12.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer+integer2);


        Map<String, Integer> rddMap1 = MapUtil.sortDescend( reduceMap1.collectAsMap());
        Map<String, Integer> rddMap2= MapUtil.sortDescend(reduceMap2.collectAsMap());
        Map<String, Integer> rddMap3 =MapUtil.sortDescend( reduceMap3.collectAsMap());
        Map<String, Integer> rddMap4 = MapUtil.sortDescend(reduceMap4.collectAsMap());
        Map<String, Integer> rddMap5 = MapUtil.sortDescend(reduceMap5.collectAsMap());
        Map<String, Integer> rddMap6 = MapUtil.sortDescend(reduceMap6.collectAsMap());
        Map<String, Integer> rddMap7 = MapUtil.sortDescend(reduceMap7.collectAsMap());
        Map<String, Integer> rddMap8 = MapUtil.sortDescend(reduceMap8.collectAsMap());
        Map<String, Integer> rddMap9 = MapUtil.sortDescend(reduceMap9.collectAsMap());
        Map<String, Integer> rddMap10 = MapUtil.sortDescend(reduceMap10.collectAsMap());
        Map<String, Integer> rddMap11 = MapUtil.sortDescend(reduceMap11.collectAsMap());
        Map<String, Integer> rddMap12 = MapUtil.sortDescend(reduceMap12.collectAsMap());



        ArrayList product1 = new ArrayList<>();
        ArrayList product2 = new ArrayList<>();
        ArrayList product3 = new ArrayList<>();
        ArrayList product4 = new ArrayList<>();
        ArrayList product5 = new ArrayList<>();
        ArrayList product6 = new ArrayList<>();
        ArrayList product7 = new ArrayList<>();
        ArrayList product8 = new ArrayList<>();
        ArrayList product9 = new ArrayList<>();
        ArrayList product10 = new ArrayList<>();
        ArrayList product11 = new ArrayList<>();
        ArrayList product12 = new ArrayList<>();

//        if (rddMap4.size() != 0) {
//            System.out.println(rddMap4.keySet().stream().iterator().next());
//            System.out.println(rddMap4.values().stream().iterator().next());
//        }else System.out.println(1);



        product1.add(0);product1.add(rddMap1.size() == 0? 0: rddMap1.values().stream().iterator().next());product1.add(rddMap1.size() ==0?"Lemon":rddMap1.keySet().stream().iterator().next());
        product2.add(0);product2.add(rddMap2.size() ==0? 0:rddMap2.values().stream().iterator().next());product2.add(rddMap2.size() ==0?"Matcha":rddMap2.keySet().stream().iterator().next());
        product3.add(0);product3.add(rddMap3.size() ==0? 0:rddMap3.values().stream().iterator().next());product3.add(rddMap3.size() ==0?"Latte":rddMap3.keySet().stream().iterator().next());
        product4.add(0);product4.add(rddMap4.size() ==0? 0:rddMap4.values().stream().iterator().next());product4.add(rddMap4.size() ==0?"Milk":rddMap4.keySet().stream().iterator().next());
        product5.add(0);product5.add(rddMap5.size() ==0?0:rddMap5.values().stream().iterator().next());product5.add(rddMap5.size() ==0?"Tea":rddMap5.keySet().stream().iterator().next());
        product6.add(0);product6.add(rddMap6.size() ==0?0:rddMap5.values().stream().iterator().next());product6.add(rddMap6.size() ==0?"Cocoa":rddMap6.keySet().stream().iterator().next());
        product7.add(0);product7.add(rddMap7.size() ==0?0:rddMap5.values().stream().iterator().next());product7.add(rddMap7.size()==0?"Brownie":rddMap7.keySet().stream().iterator().next());
        product8.add(0);product8.add(rddMap8.size() ==0?0:rddMap5.values().stream().iterator().next());product8.add(rddMap8.size()==0?"Juice":rddMap8.keySet().stream().iterator().next());
        product9.add(0);product9.add(rddMap9.size() ==0?0:rddMap5.values().stream().iterator().next());product9.add(rddMap9.size()==0?"Brownie":rddMap9.keySet().stream().iterator().next());
        product10.add(0);product10.add(rddMap10.size() ==0?0:rddMap5.values().stream().iterator().next());product10.add(rddMap10.size()==0?"apple":rddMap10.keySet().stream().iterator().next());
        product11.add(0);product11.add(rddMap11.size() ==0?0:rddMap5.values().stream().iterator().next());product11.add(rddMap11.size() ==0?"banana":rddMap11.keySet().stream().iterator().next());
        product12.add(0);product12.add(rddMap12.size() ==0?0:rddMap5.values().stream().iterator().next());product12.add(rddMap12.size() ==0?"orange":rddMap12.keySet().stream().iterator().next());


//        System.out.println(product1);
//        System.out.println(product4);

        ArrayList head = new ArrayList();   //['score', 'amount', 'product'],
        head.add("score");head.add("amount");head.add("product");

        ArrayList returnData = new ArrayList();

        returnData.add(head);returnData.add(product1);returnData.add(product2);returnData.add(product3);returnData.add(product4);returnData.add(product5);returnData.add(product6);returnData.add(product7);
        returnData.add(product8);returnData.add(product9);returnData.add(product10);returnData.add(product11);returnData.add(product12);


        return returnData;
    }
}
